package com.example.mqtt.provider.impl;

import com.example.mqtt.doman.consts.MqttChannelName;
import com.example.mqtt.provider.ChannelMessageHandler;
import com.example.mqtt.provider.ChannelTopicMessageHandler;
import com.example.mqtt.utils.ApplicationContextUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.annotation.DependsOn;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.stereotype.Component;
import org.springframework.util.AntPathMatcher;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.stream.Collectors;

@Slf4j
@Component
@DependsOn("applicationContextUtil")
public class DefaultChannelMessageHandler implements ChannelMessageHandler, InitializingBean {
    private final AntPathMatcher antPathMatcher = new AntPathMatcher();

    private Map<String,LinkedHashSet<ChannelTopicMessageHandler>> channelTopicMessageHandlers = new HashMap<>();


    @Override
    public String boundChannel() {
        return MqttChannelName.INBOUND_DEFAULT;
    }




    @Override
    public void handleMessage(Message<?> message) throws MessagingException {
        String topic =String.valueOf(message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC));
        if (channelTopicMessageHandlers.isEmpty()) {
            log.warn("没有主题消息处理实现无法处理:{}",topic);
            return;
        }
        Set<ChannelTopicMessageHandler> handlerList = channelTopicMessageHandlers.get(topic);
        // 如果没有明确指定,尝试匹配
        if (CollectionUtils.isEmpty(handlerList)) {
            // 没有找到尝试匹配主题里面有通配符，且匹配的主题
            List<String> collect = channelTopicMessageHandlers.keySet().stream().filter(ele -> ele.contains("*") && antPathMatcher.match(ele, topic)).collect(Collectors.toList());
            if (collect.isEmpty()) {
                log.warn("未查询到主题:{},的消息处理",topic);
                return;
            }
            if (collect.size()>1) {
                log.error("匹配到多个满足条件的主题:{},的消息处理",topic);
                return;
            }
            handlerList = channelTopicMessageHandlers.get(collect.get(0));
        }
        // 消息处理
        for (ChannelTopicMessageHandler handler : handlerList) {
            String topicPattern = handler.topicPattern();
            if (antPathMatcher.match(topicPattern,topic)) {
                handler.handleMessage(message);
            }
        }
    }


    @Override
    public void afterPropertiesSet() throws Exception {
        // 被调用两次，这里做个非空判断保证只执行一次
        if (channelTopicMessageHandlers.isEmpty()) {
            Map<String, ChannelTopicMessageHandler> beans = ApplicationContextUtil.getBeans(ChannelTopicMessageHandler.class);
            List<ChannelTopicMessageHandler> collect = beans.values().stream().filter(ele -> boundChannel().equals(ele.boundChannel())).sorted(Comparator.comparing(ChannelTopicMessageHandler::order)).collect(Collectors.toList());
            for (ChannelTopicMessageHandler handler: collect) {
                String topicPattern = handler.topicPattern();
                Set<ChannelTopicMessageHandler> handlerList = channelTopicMessageHandlers.computeIfAbsent(topicPattern, key -> new LinkedHashSet<>());
                handlerList.add(handler);
            }
        }
    }
}
